﻿#include "threadpool.h"

#include <functional>
#include <thread>
#include <iostream>

const int TASK_MAX_THRESHHODE = INT32_MAX;
const int THREAD_MAX_THRESHHODE = 512;
const int THREAD_MAX_IDEL_TIME = 60;  // 单位：秒

// 线程池构造
ThreadPool::ThreadPool()
	: initThreadSize_(0)
	, taskSize_(0)
	, idleThreadSize_(0)
	, curThreadSize_(0)
	, threadSizeThreshHold_(THREAD_MAX_THRESHHODE)
	, taskQueMaxThreshHold_(TASK_MAX_THRESHHODE)
	, poolMode_(PoolMode::MODE_FIXED)
	, isPoolRunning_(false)
{}

// C++编程规范中：只要显式写构造，就要显式写析构，功能要闭环，
// 即使现在析构中什么都不干，难免以后要向其中再添加东西，可能要析构
// 线程池析构
ThreadPool::~ThreadPool() 
{
	isPoolRunning_ = false;

	// 等待线程池中所有的线程返回
	std::unique_lock<std::mutex> lock(taskQueMtx_);
	notEmpty_.notify_all();
	exitCond_.wait(lock, [&]() ->bool { return threads_.size() == 0; });
	
}

// 设置线程池模式
void ThreadPool::setMode(PoolMode mode) 
{
	if (checkRunningState())
		return;

	poolMode_ = mode;
}

// 设置线程池任务队列最大数量
void ThreadPool::setTaskQueMaxThreshHold(int threshhold)
{
	if (checkRunningState())
		return;

	taskQueMaxThreshHold_ = threshhold;
}

// 给线程池提交任务
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	// 获取锁
	std::unique_lock<std::mutex> lock(taskQueMtx_);

	// 线程通信
	//while (taskQue_.size() == taskQueMaxThreshHold_)
	//{
	//	notFull_.wait(lock); // 阻塞等待
	//	// 如果阻塞了1s还没等到，直接返回，提示用户任务队列已满，稍后操作
	//}

	// notFull_.wait(lock, [&]() ->bool { return taskQue_.size() < taskQueMaxThreshHold_; });
	if (! notFull_.wait_for(lock, std::chrono::seconds(1),
		[&]() ->bool { return taskQue_.size() < taskQueMaxThreshHold_; }) )
	{
		std::cerr << "task queue is full, submit task fail." << std::endl;
		// return task->getReuslt();  // Task  Result  线程执行完Task任务，Task就被析构掉了
		return Result(sp, false);
	}

	// 如果有空余，把任务放入任务队列中
	taskQue_.emplace(sp);
	taskSize_++;

	// 因为放了新任务，任务队列一定不空了，在notEmpty_上通知，分配线程执行任务
	notEmpty_.notify_all();

	// cached模式，任务处理比较紧急，场景：小而快
	if (poolMode_ == PoolMode::MODE_CACHED 
		&& taskSize_ > idleThreadSize_ 
		&& curThreadSize_ < threadSizeThreshHold_)
	{
		// 创建新线程
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		// threads_.emplace_back(std::move(ptr));
		int threadId = ptr->getId();
		threads_.emplace(threadId, std::move(ptr));
		// 启动线程
		threads_[threadId]->start();
		// 修改相关变量
		curThreadSize_++;	
		idleThreadSize_++;

		std::cout << ">>> create new thread successful" << std::endl;
	}

	// 返回任务的Result对象
	return Result(sp);
}

// 开启线程池
void ThreadPool::start(int initThreadSize)
{
	isPoolRunning_ = true;

	// 记录初始线程数量
	initThreadSize_ = initThreadSize;
	curThreadSize_ = initThreadSize;	

	// 创建线程对象
	for (int i = 0; i < initThreadSize; ++i)
	{
		// 创建新线程
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		// threads_.emplace_back(std::move(ptr));
		int threadId = ptr->getId();
		threads_.emplace(threadId, std::move(ptr));
	}

	// 启动所有线程
	for (int i = 0; i < initThreadSize; ++i) 
	{
		threads_[i]->start(); // 去执行线程函数
		idleThreadSize_++;    // 记录初始空闲线程的数量
	}

	// isPoolRunning_ = true;
}

// 定义线程函数
void ThreadPool::threadFunc(int threadId)
{
	auto lastTime = std::chrono::steady_clock().now();
	for (;;)
	{
		std::shared_ptr<Task> task = nullptr; 
		{
			// 获取锁
			std::unique_lock<std::mutex> lock(taskQueMtx_);
			std::cout << std::this_thread::get_id() << "尝试获取任务" << std::endl;

			// 超过initThreadSize_的线程要进行回收
			
			while (taskQue_.size() == 0)
			{
				//// 线程池要回收线程资源
				if (!isPoolRunning_)
				{
					threads_.erase(threadId);
					curThreadSize_--;
					idleThreadSize_--;
					std::cout << std::this_thread::get_id() << " exit!" << std::endl;
					exitCond_.notify_all();
					return;
				}
				if (poolMode_ == PoolMode::MODE_CACHED)
				{
					// 条件变量超时返回
					if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						auto now = std::chrono::steady_clock().now();
						auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						if (dur.count() >= THREAD_MAX_IDEL_TIME && curThreadSize_ > initThreadSize_)
						{
							// 回收当前线程
							// 修改记录线程相关的变量的值
							// 把线程对象从线程列表容器中删除
							threads_.erase(threadId);
							curThreadSize_--;
							idleThreadSize_--;
							// 问题：ThreadFunc 是被哪个 Thread执行   
							std::cout << std::this_thread::get_id() << " exit!" << std::endl;
							return;
						}
					}
				}
				else
				{
					// 等待 notEmpty 条件
					notEmpty_.wait(lock);
				}
			}
		
			idleThreadSize_--;

			// 从任务队列中取一个任务出来
			task = taskQue_.front();
			taskQue_.pop();
			taskSize_--;
			
			std::cout << std::this_thread::get_id() << "获取任务成功" << std::endl;

			// 如果依然有剩余任务，继续通知其他线程执行任务
			if (taskQue_.size() > 0)
			{
				notEmpty_.notify_all();
			}

			// 取出一个任务，进行通知，通知可以继续生产任务
			notFull_.notify_all();
		} // 保证释放锁之后再执行任务

		// 当前线程执行这个任务
		if (task != nullptr)
		{
			// task->run(); // 执行任务，把任务的返回值通过setVal接口返回给Result
			task->exec();
		}
		lastTime = std::chrono::steady_clock().now();
		idleThreadSize_++; 
	}
}

// 设置线程数量的上限
void ThreadPool::setThreadMaxThreshHold(int threshhold)
{
	if (checkRunningState())
		return;
	if (this->poolMode_ != PoolMode::MODE_CACHED)
		std::cout << "非Cached模式不可设置threadSizeThreshHold_" << std::endl;
	threadSizeThreshHold_ = threshhold;
}

// 检查线程池的运行状态
bool ThreadPool::checkRunningState() const
{
	return isPoolRunning_;
}





// =====================  Task方法实现
Task::Task()
	: result_(nullptr)
{}

void Task::exec()
{
	if (result_ != nullptr)
		result_->setVal(run()); // 这里发生多态调用
}
void Task::setReuslt(Result* result)
{
	result_ = result;
}
// =====================  线程方法实现

// 启动线程
void Thread::start() 
{
	// 要执行一个线程函数
	std::thread t(func_, threadId_);
	t.detach();
}

// 线程函数要是写在线程类中，不太合适，线程要访问的相关变量都是线程池中


int Thread::generateId_ = 0;
// 线程构造
Thread::Thread(ThreadFunc func)
	: func_(func)
	, threadId_(generateId_ ++)
{}

// 线程析构
Thread::~Thread()
{
	std::cout << std::this_thread::get_id()  << " ~Thread()" << std::endl;
}

// 获得线程ID
int Thread::getId() const
{
	return threadId_;
}


// ======================= Result 方法实现
Result::Result(std::shared_ptr<Task> task, bool isVaild)
	: isVaild_(isVaild)
	, task_(task)
{
	task_->setReuslt(this);
}

Any Result::get()
{
	if (!isVaild_)
		return "";

	sem_.wait();
	return std::move(any_);
}

void Result::setVal(Any any)
{
	this->any_ = std::move(any);
	sem_.post(); // 已经获得的任务的返回值 增加信号量资源
}